﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using NUnit.Framework;
using System.Threading;

namespace Flower.Actors.Tests
{
    [TestFixture]
    public class ActorTests
    {
        private class MessageComparer : IComparer<IMessage<ActorTests>>
        {
            public int Compare(IMessage<ActorTests> x, IMessage<ActorTests> y)
            {
                return 0;
            }
        }

        private class InfiniteLoopMessage : IMessage<ActorTests>
        {
            public void Process(ActorTests processor)
            {
                InfiniteLoop();
            }
        }

        private class ProviderMessage : IMessage<ActorTests>
        {
            public BlockingCollection<object> Result;

            public void Process(ActorTests processor)
            {
                Result.Add(null);
            }
        }

        private class CounterMessage : IMessage<ActorTests>
        {
            public int Count = 0;

            public void Process(ActorTests processor)
            {
                Count++;
            }
        }

        private class SuspendMessage : IMessage<ActorTests>
        {
            private bool _thrown;

            public void Process(ActorTests processor)
            {
                if (!_thrown)
                {
                    _thrown = true;
                    throw new ActorSuspendException();
                }
            }
        }

        private class ExceptionMessage : IMessage<ActorTests>
        {
            public void Process(ActorTests processor)
            {
                throw new Exception();
            }
        }

        private class WaitMessage : IMessage<ActorTests>
        {
            private readonly TimeSpan _delay;

            public WaitMessage(TimeSpan delay)
            {
                _delay = delay;
            }

            public void Process(ActorTests processor)
            {
                Thread.Sleep(_delay);
            }
        }

        [Test]
        public void ActorTimeout()
        {
            var actor = new Actor<ActorTests, IMessage<ActorTests>>(this, TimeSpan.FromSeconds(1));

            actor.Send(new InfiniteLoopMessage());

            var result = new BlockingCollection<object>();
            actor.Send(new ProviderMessage { Result = result });
            result.Take();
        }

        [Test]
        public void ActorSuspension()
        {
            var actor = new Actor<ActorTests, IMessage<ActorTests>>(this, TimeSpan.FromSeconds(1));

            var cnt = new CounterMessage();

            actor.Send(cnt);
            actor.Send(cnt);
            Thread.Sleep(100);

            Assert.AreEqual(2, cnt.Count);

            actor.Send(new SuspendMessage());

            actor.Send(cnt);
            actor.Send(cnt);
            Thread.Sleep(100);

            Assert.AreEqual(2, cnt.Count);

            actor.Resume();
            Thread.Sleep(100);

            Assert.AreEqual(4, cnt.Count);
        }

        private static void InfiniteLoop()
        {
            while (true) {}
        }

        [Test]
        public void ErrorHandling()
        {
            IActor<IMessage<ActorTests>> actor =
                new MultiActor<ActorTests, IMessage<ActorTests>>
                (
                    10, 
                    i => new Actor<ActorTests, IMessage<ActorTests>>(this, TimeSpan.FromSeconds(1))
                );

            var errors = new BlockingCollection<int>();

            actor.Error += 
                (sender, args) => errors.Add(0);

            const int msgCount = 10;

            for (int i = 0; i < msgCount; i++)
            {
                actor.Send(new ExceptionMessage());
            }

            for (int i = 0; i < msgCount; i++)
            {
                int itm;

                if (!errors.TryTake(out itm, TimeSpan.FromSeconds(10)))
                {
                    Assert.Fail("Timeout.");
                }
            }
        }

        [Test]
        public void WaitAllMessages()
        {
            const int msgCount = 10;
            IActor<IMessage<ActorTests>> actor =
                new MultiActor<ActorTests, IMessage<ActorTests>>
                (
                    msgCount,
                    i => new Actor<ActorTests, IMessage<ActorTests>>(this)
                );

            for (int i = 0; i < msgCount; i++)
            {
                actor.Send(new WaitMessage(TimeSpan.FromMilliseconds(300)));
            }

            Assert.IsTrue(actor.WaitAllMessages(null));

            for (int i = 0; i < msgCount; i++)
            {
                actor.Send(new WaitMessage(TimeSpan.FromMilliseconds(300)));
            }

            Assert.IsFalse(actor.WaitAllMessages(TimeSpan.FromMilliseconds(100)));
            Assert.IsTrue(actor.WaitAllMessages(null));
        }

        private class IntComparer : IComparer<int>
        {
            public int Compare(int x, int y)
            {
                return x - y;
            }
        }

        [Test]
        public void PriorityQueueWorkStealing()
        {
            var queue = new PriorityQueue<int>(new IntComparer());

            Assert.IsTrue(queue.TryAdd(1));

            Assert.AreEqual(0, queue.Steal(0.5).Length);
            Assert.AreEqual(1, queue.Count);

            Assert.IsTrue(queue.TryAdd(1));
            Assert.IsTrue(queue.TryAdd(1));
            Assert.IsTrue(queue.TryAdd(1));

            Assert.AreEqual(2, queue.Steal(0.5).Length);
            Assert.AreEqual(2, queue.Count);
        }

        [Test]
        public void MultiActorRebalancing()
        {
            const int queuesCount = 3;
            IActor<IMessage<ActorTests>> actor =
                new MultiActor<ActorTests, IMessage<ActorTests>>
                (
                    queuesCount,
                    i => 
                        new Actor<ActorTests, IMessage<ActorTests>>
                        (
                            this, 
                            new PriorityQueue<IMessage<ActorTests>>(new MessageComparer())
                        )
                );

            Thread.Sleep(TimeSpan.FromSeconds(1));

            var started = DateTime.Now;

            actor.Send(new WaitMessage(TimeSpan.FromSeconds(1.5)));

            for (int i = 0; i < 9; i++)
            {
                actor.Send(new WaitMessage(TimeSpan.FromSeconds(0.33)));
            }

            actor.WaitAllMessages(TimeSpan.FromSeconds(3));

            Assert.IsTrue(DateTime.Now - started < TimeSpan.FromSeconds(2));
        }

        [Test]
        public void MultiActorNoRebalancing()
        {
            const int queuesCount = 3;
            IActor<IMessage<ActorTests>> actor =
                new MultiActor<ActorTests, IMessage<ActorTests>>
                (
                    queuesCount,
                    i => new Actor<ActorTests, IMessage<ActorTests>>(this)
                );

            Thread.Sleep(TimeSpan.FromSeconds(1));

            var started = DateTime.Now;

            actor.Send(new WaitMessage(TimeSpan.FromSeconds(1.5)));

            for (int i = 0; i < 9; i++)
            {
                actor.Send(new WaitMessage(TimeSpan.FromSeconds(0.33)));
            }

            actor.WaitAllMessages(TimeSpan.FromSeconds(10));

            Assert.IsTrue(DateTime.Now - started > TimeSpan.FromSeconds(2));
        }
    }
}
